package com.atguigu.edu.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.base.BaseApp;
import com.atguigu.edu.realtime.common.bean.UserAnalyzeBean;
import com.atguigu.edu.realtime.common.constant.Constant;
import com.atguigu.edu.realtime.common.function.BeanToJsonStrMapFunction;
import com.atguigu.edu.realtime.common.util.DateFormatUtil;
import com.atguigu.edu.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SideOutputDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;

public class DwsUserUserActionAnalyzeWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsUserUserActionAnalyzeWindow().start(
                10026,
                4,
                "dws_user_user_action_analyze_window",
                Constant.TOPIC_DWD_TRAFFIC_PAGE
        );
    }
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.map(
                new MapFunction<String, JSONObject>() {
                    @Override
                    public JSONObject map(String value) throws Exception {
                        return JSON.parseObject(value);
                    }
                }
        );

        SingleOutputStreamOperator<UserAnalyzeBean> beanDS = jsonObjDS.process(
                new ProcessFunction<JSONObject, UserAnalyzeBean>() {
                    @Override
                    public void processElement(JSONObject value, ProcessFunction<JSONObject, UserAnalyzeBean>.Context ctx, Collector<UserAnalyzeBean> out) throws Exception {
                        String pageId = value.getJSONObject("page").getString("page_id");
                        Long ts = value.getLong("ts");
                        UserAnalyzeBean userAnalyzeBean = new UserAnalyzeBean("", "", "", 0L, 0L, 0L, 0L, 0L, ts);
                        if ("home".equals(pageId)) {
                            userAnalyzeBean.setHomeViewCt(1L);
                        } else if ("course_list".equals(pageId)) {
                            userAnalyzeBean.setCourseDetailCt(1L);
                        } else if ("cart".equals(pageId)) {
                            userAnalyzeBean.setCartCt(1L);
                        } else if ("order".equals(pageId)) {
                            userAnalyzeBean.setOrderCt(1L);
                        } else if ("payment".equals(pageId)) {
                            userAnalyzeBean.setPayCt(1L);
                        }
                        out.collect(userAnalyzeBean);
                    }
                }
        );

        SingleOutputStreamOperator<UserAnalyzeBean> andWatermarks = beanDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<UserAnalyzeBean>forBoundedOutOfOrderness(Duration.ZERO)
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<UserAnalyzeBean>() {
                                    @Override
                                    public long extractTimestamp(UserAnalyzeBean element, long recordTimestamp) {
                                        return element.getTs();
                                    }
                                }
                        )
        );
        AllWindowedStream<UserAnalyzeBean, TimeWindow> windowedStream = andWatermarks.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(10)));

        SingleOutputStreamOperator<UserAnalyzeBean> reduceDS = windowedStream.reduce(
                new ReduceFunction<UserAnalyzeBean>() {
                    @Override
                    public UserAnalyzeBean reduce(UserAnalyzeBean value1, UserAnalyzeBean value2) throws Exception {
                        value1.setHomeViewCt(value1.getHomeViewCt() + value2.getHomeViewCt());
                        value1.setCourseDetailCt(value1.getCourseDetailCt() + value2.getCourseDetailCt());
                        value1.setCartCt(value1.getCartCt() + value2.getCartCt());
                        value1.setOrderCt(value1.getOrderCt() + value2.getOrderCt());
                        value1.setPayCt(value1.getPayCt() + value2.getPayCt());
                        return value1;
                    }
                },
                new AllWindowFunction<UserAnalyzeBean, UserAnalyzeBean, TimeWindow>() {
                    @Override
                    public void apply(TimeWindow window, Iterable<UserAnalyzeBean> values, Collector<UserAnalyzeBean> out) throws Exception {
                        UserAnalyzeBean userAnalyzeBean = values.iterator().next();
                        String stt = DateFormatUtil.tsToDateTime(window.getStart());
                        String edt = DateFormatUtil.tsToDateTime(window.getEnd());
                        String curdate = DateFormatUtil.tsToDate(window.getStart());
                        userAnalyzeBean.setStt(stt);
                        userAnalyzeBean.setEdt(edt);
                        userAnalyzeBean.setCurDate(curdate);
                        out.collect(userAnalyzeBean);
                    }
                }
        );
        reduceDS.print();
        reduceDS.map(new BeanToJsonStrMapFunction<UserAnalyzeBean>())
                .sinkTo(FlinkSinkUtil.getDorisSink("dws_user_user_action_analyze_window"));


    }
}
